package org.javaboy.redis_stream_demo.service;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import org.javaboy.redis_stream_demo.config.RedisQueueConfig;
import org.javaboy.redis_stream_demo.model.User;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.connection.stream.StreamReadOptions;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.Date;
import java.util.List;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 消费者
 */
@Component
public class StreamConsumer {
    @Autowired
    RedisTemplate redisTemplate;

    //消息消费的线程池，据此可以并发消费
    ThreadPoolExecutor executor;
    //消息消费的开关
    boolean stop = false;

    @PostConstruct
    public void init() {
        executor = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS, new LinkedBlockingDeque<>(), runnable -> {
            Thread thread = new Thread(runnable);
            thread.setName("consumer");
            thread.setDaemon(true);
            return thread;
        });
        StreamReadOptions streamReadOptions = StreamReadOptions.empty()
                //阻塞时间，0 代表一直阻塞，这个地方的值不能超过 spring.data.redis.timeout
                .block(Duration.ofSeconds(10))
                //每次拉取的消息数量
                .count(10);
        //消息的偏移量，其实就是消息 ID，这里表示从哪个消息开始拉取
        //0-0 表示从第一个消息开始拉取
        StringBuilder readOffset = new StringBuilder("0-0");
        executor.execute(() -> {
            while (!stop) {
                //拉取消息
                List<ObjectRecord<String, User>> list = redisTemplate.opsForStream().read(User.class, streamReadOptions,
                        StreamOffset.create(RedisQueueConfig.QUEUE_NAME, ReadOffset.from(readOffset.toString())));
                if (list.isEmpty()) {
                    System.out.println(new Date() + ">>>" + Thread.currentThread().getName() + " 拉取消息为空");
                    continue;
                }
                for (ObjectRecord<String, User> record : list) {
                    System.out.println(Thread.currentThread().getName() + " 拉取的消息 ID 为：" + record.getId() + " ；消息内容为：" + record.getValue());
                    readOffset.setLength(0);
                    //这是最新的 id，下次拉取消息时，从这里开始拉取
                    readOffset.append(record.getId());
                    //可以手动删除消息（根据业务来定）
                    redisTemplate.opsForStream().delete(RedisQueueConfig.QUEUE_NAME, record.getId());
                }
            }
        });
    }

    @PreDestroy
    public void stop() {
        stop = true;
        executor.shutdown();
    }

}
